// =============================================================
// space test btw. server/client
// ---------------------------------------
// Author : Woo-Hyun Kim
// Email : woorung@icu.ac.kr or woorung@nhncorp.com
// Date : Oct. 8, 2007
// =============================================================
#include <iostream>

#include <sys/timeb.h>
#include <time.h>

#include "common_coord.h"
#include "property/CProperty.h"

#include "thread/CThread.h"
#include "service/CServiceRepository.h"

#include "CKeepAliveThread.h"

#include "CDiscoveryService.h"
#include "CDiscoveryProxy.h"

#include "md5/CMD5.h"

#include "CSpaceService.h"

#include "CWorkSpaces.h"

int write(CChord *chord, pair<string,string> &kv)
{
	char *key = (char *)kv.first.c_str();
	char *value = (char *)kv.second.c_str();

	char *cmd = CUBE_SPACE_WRITE;
	CObjectArray *pobjs = new CObjectArray(3);
	(*pobjs)[0] = new cube__CKeyValue();
	dynamic_cast<cube__CKeyValue*>((*pobjs)[0])->setKey(CUBE_SPACE_ACTION);
	dynamic_cast<cube__CKeyValue*>((*pobjs)[0])->setValue(cmd);
	// the end of parameters
	(*pobjs)[1] = NULL;

	(*pobjs)[2] = new cube__CKeyValue();
	dynamic_cast<cube__CKeyValue *>((*pobjs)[2])->set(key, value);

#ifdef DEBUG
	cout << "inserting key = " << key << endl;
	cout << "inserting value = " << value << endl;
#endif

	// find a specific space responsible for the data key
	string space_key = key;
	CServiceInfo found_space = chord->lookup((char *)space_key.c_str());

#ifdef DEBUG
	cout << "####################################################" << endl;
	cout << "service name = " << found_space.getName() << endl;
	cout << "host ip = " << found_space.getIP() << endl;
	cout << "host port = " << found_space.getPort() << endl;
	cout << "####################################################" << endl;
#endif

	CServiceProxy space_proxy;
	space_proxy.connect(found_space.getIP(), found_space.getPort());

	cube__CService space_service;
	space_service.service(found_space.getName());
	//space_service.instance("sample.space.service");

	struct cube__invokeResponse res;

	if(space_proxy.invoke(&space_service, pobjs, res)!=SOAP_OK) {
		soap_print_fault(space_proxy.soap(), stderr);

		delete (*pobjs)[0];
		// delete (*pobjs)[1]; // 1 is NULL
		delete (*pobjs)[2];
		delete pobjs;
	}

	delete (*pobjs)[0];
	// delete (*pobjs)[1]; // 1 is NULL
	delete (*pobjs)[2];
	delete pobjs;
}

int read(CChord *chord, pair<string,string> &kv)
{
	char *cmd = CUBE_SPACE_READ_IF_EXISTS;

	//if(wait) cmd = "read";
	//else cmd = "readIfExists";

	CObjectArray *pobjs = new CObjectArray(3);
	(*pobjs)[0] = new cube__CKeyValue();
	dynamic_cast<cube__CKeyValue *>((*pobjs)[0])->setKey(CUBE_SPACE_ACTION);
	dynamic_cast<cube__CKeyValue *>((*pobjs)[0])->setValue(cmd);
	// the end of parameters
	(*pobjs)[1] = NULL;

	if(kv.first=="") (*pobjs)[2] = NULL;
	else {
		(*pobjs)[2] = new cube__CKeyValue();
		dynamic_cast<cube__CKeyValue *>((*pobjs)[2])->setKey((char *)kv.first.c_str());
	}

	string space_key = kv.first;

	// find a specific space responsible for the data key
	CServiceInfo found_space = chord->lookup((char *)space_key.c_str());

	// create a proxy for the space
	CServiceProxy space_proxy;
	space_proxy.connect(found_space.getIP(), found_space.getPort());

	cube__CService space_service;
	space_service.service(found_space.getName());
	//space_service.instance("sample.space.service");

	struct cube__invokeResponse res;

	if(space_proxy.invoke(&space_service, pobjs, res)!=SOAP_OK) {
		soap_print_fault(space_proxy.soap(), stderr);

		delete (*pobjs)[0];
		// delete (*pobjs)[1]; // 1 is NULL
		delete (*pobjs)[2];
		delete pobjs;

		return -1;
	}

	CObjectArray *array = dynamic_cast<CObjectArray *>(res.out);
	if(array!=NULL) {

		cube__CKeyValue *keyval = dynamic_cast<cube__CKeyValue *>(array->__ptr[0]);
		if(keyval==NULL) {
			cout << "not key-value type" << endl;
			delete (*pobjs)[0];
			// delete (*pobjs)[1]; // 1 is NULL
			delete (*pobjs)[2];
			delete pobjs;

			return -1;
		}

#ifdef DEBUG
		cout << "found key = " << keyval->getKey() << endl;
		cout << "found value = " << keyval->getValue() << endl;
#endif

		kv.first = keyval->getKey();
		kv.second = keyval->getValue();
	}
	else {
		delete (*pobjs)[0];
		// delete (*pobjs)[1]; // 1 is NULL
		delete (*pobjs)[2];
		delete pobjs;
		return -1;
	}

	delete (*pobjs)[0];
	// delete (*pobjs)[1]; // 1 is NULL
	delete (*pobjs)[2];
	delete pobjs;

	return 1;
}

int take(CChord *chord, pair<string,string> &kv)
{
	char *cmd = CUBE_SPACE_TAKE_IF_EXISTS;

	CObjectArray *pobjs = new CObjectArray(3);
	(*pobjs)[0] = new cube__CKeyValue();
	dynamic_cast<cube__CKeyValue *>((*pobjs)[0])->setKey(CUBE_SPACE_ACTION);
	dynamic_cast<cube__CKeyValue *>((*pobjs)[0])->setValue(cmd);
	// the end of parameters
	(*pobjs)[1] = NULL;

	
	// take any key
	(*pobjs)[2] = NULL;
/*
	if(kv.first=="") (*pobjs)[2] = NULL;
	else {
		(*pobjs)[2] = new cube__CKeyValue();
		dynamic_cast<cube__CKeyValue *>((*pobjs)[2])->setKey((char *)kv.first.c_str());
	}
*/

	string space_key = kv.first;

	// find a specific space responsible for the data key
	CServiceInfo found_space = chord->lookup((char *)space_key.c_str());

	// create a proxy for the space
	CServiceProxy space_proxy;
	space_proxy.connect(found_space.getIP(), found_space.getPort());

	cube__CService space_service;
	space_service.service(found_space.getName());
	//space_service.instance("sample.space.service");

	struct cube__invokeResponse res;

	if(space_proxy.invoke(&space_service, pobjs, res)!=SOAP_OK) {
		soap_print_fault(space_proxy.soap(), stderr);

		delete (*pobjs)[0];
		// delete (*pobjs)[1]; // 1 is NULL
		delete (*pobjs)[2];
		delete pobjs;

		return -1;
	}

	CObjectArray *array = dynamic_cast<CObjectArray *>(res.out);
	if(array!=NULL) {
		cube__CKeyValue *keyval = dynamic_cast<cube__CKeyValue *>(array->__ptr[0]);
		if(keyval==NULL) {
			cout << "not key-value type" << endl;
			delete (*pobjs)[0];
			// delete (*pobjs)[1]; // 1 is NULL
			delete (*pobjs)[2];
			delete pobjs;

			return -1;
		}

#ifdef DEBUG
		cout << "found key = " << keyval->getKey() << endl;
		cout << "found value = " << keyval->getValue() << endl;
#endif

		kv.first = keyval->getKey();
		kv.second = keyval->getValue();
	}
	else {
		cout << "###########################################" << endl;
		cout << "no no no" << endl;
		cout << "###########################################" << endl;

		delete (*pobjs)[0];
		// delete (*pobjs)[1]; // 1 is NULL
		delete (*pobjs)[2];
		delete pobjs;
		return -1;
	}

	delete (*pobjs)[0];
	// delete (*pobjs)[1]; // 1 is NULL
	delete (*pobjs)[2];
	delete pobjs;

	return 1;
}

void randomize(int seed)
{
	srand(seed);
}

int new_random()
{ 
	return rand();
}

string &simple_hash(char *key, string &hashkey)
{
        CMD5 md5;
        unsigned char digest[32];
        md5.GetMD5Message(key, digest);

        char tmp[256];
        hashkey = "";
        for(int i=0;i<16;i++) {
                sprintf(tmp, "%02x",digest[i]);
                hashkey += tmp;
        }

        return hashkey;
}

pair<string,string> &autoGenKV(pair<string,string> &kv)
{
	int random_key = new_random();

	char dd[100];
	sprintf(dd, "%d", random_key);
	kv.first = dd;
	kv.second = "";
	
	char data[100];
	for(int i=0;i<10;i++) {
		sprintf(data, "%s%d", (char *)kv.first.c_str(), i);
		string gen_value = "";
		gen_value = simple_hash(data, gen_value);

		kv.second += gen_value;
	}

	return kv;
}

// cube discovery
// cube discovery server
void cube_sd(int port)
{
	CServiceRepository *service_container = NULL;
	service_container = new CServiceRepository(port);

	// discovery
	CDiscoveryService discovery_service(CUBE_SERVICE_DISCOVERY);
	service_container->plugin(&discovery_service);

	cout << "===================================================" << endl;
	cout << COORD_VER << " discovery server for space test" <<  endl;
	cout << endl;
	cout << "# available services" << endl;
	cout << discovery_service.getName() << endl;
	cout << "===================================================" << endl;

	service_container->run();

	delete service_container;
}

// cube itself
void cube_sp(char *host, int port)
{
	CServiceRepository *service_container = NULL;

	randomize(getpid());

	// set random port
	int random_port = (new_random()%1000)+10080;
	service_container = new CServiceRepository(random_port);

	// sample space 
        CSpaceService space_service((char *)CUBE_SERVICE_SPACE);
        service_container->plugin(&space_service);

	cout << "===================================================" << endl;
	cout << COORD_VER << " space service for space test" << endl;
	cout << "run on port = " << random_port << endl;
	cout << endl;
	cout << "# available services" << endl;
	cout << space_service.getName() << endl;
	cout << "===================================================" << endl;

	// run keep-alive thread
	char thr_name[50];
	sprintf(thr_name,"thread-%d",new_random()%5000);
	CKeepAliveThread keepalive(thr_name);
	keepalive.bind(host, port, service_container);
	keepalive.keepalive((new_random()%15)+5);
	keepalive.start();

	// run cube itself
	service_container->run();
	
	delete service_container;
}

// space proxy
void cube_sp_test(char *host, int port, int test)
{
	CChord *chord = new CChord(host, port);
	chord->configure(CUBE_SERVICE_SPACE);

	randomize(time(NULL));

	int partial_count = 0;
	int partial_limit = 10000;
	bool change = false;

	while(1) {
		pair<string,string> keyval;
		keyval = autoGenKV(keyval);

		if(test==1) {
			cout << "writing (" << keyval.first << "," << keyval.second << ")" << endl;
			write(chord, keyval);
			cout << "--------- sucess" << endl;
			read(chord, keyval);
			cout << "reading (" << keyval.first << "," << keyval.second << ")" << endl;
			take(chord, keyval);
			cout << "taking (" << keyval.first << "," << keyval.second << ")" << endl;
		}
		else if(test==2) {
			if(!change) {
				cout << "writing (" << keyval.first << "," << keyval.second << ")" << endl;
				write(chord, keyval);
				read(chord, keyval);
				cout << "reading (" << keyval.first << "," << keyval.second << ")" << endl;

				partial_count++;
			}
			else {
				if(take(chord, keyval)>0) cout << "taking (" << keyval.first << "," << keyval.second << ")" << endl;

				partial_count--;
			}

			if(partial_count>=partial_limit) change = true;
			if(partial_count<=0) change = false;
		}
		else if(test==3) {
		}
	}

	delete chord;
}

// space proxy
void cube_sp_test2(char *host, int port, int test)
{
	CWorkSpaces *wspace = new CWorkSpaces(host, port);

	randomize(time(NULL));

	int partial_count = 0;
	int partial_limit = 10000;
	bool change = false;

	int bulk_limit = 10000;
	bool flush = false;
	bool suck = false;

	wspace->bulk(bulk_limit);

	while(1) {
		pair<string,string> keyval;
		keyval = autoGenKV(keyval);

		if(test==1) {
			cout << "writing (" << keyval.first << "," << keyval.second << ")" << endl;
			wspace->write(keyval);
			wspace->readIfExists(keyval);
			cout << "reading (" << keyval.first << "," << keyval.second << ")" << endl;
			wspace->takeIfExists(keyval);
			cout << "taking (" << keyval.first << "," << keyval.second << ")" << endl;
		}
		else if(test==2) {
			if(!change) {
				cout << "writing (" << keyval.first << "," << keyval.second << ")" << endl;
				wspace->write(keyval);
				wspace->readIfExists(keyval);
				cout << "reading (" << keyval.first << "," << keyval.second << ")" << endl;

				partial_count++;
			}
			else {
				// take any key in local spaces
				keyval.first = "";

				while(wspace->takeIfExists(keyval)<1);

				cout << "taking (" << keyval.first << "," << keyval.second << ")" << endl;

				partial_count--;
			}

			if(partial_count>=partial_limit) change = true;
			if(partial_count<=0) change = false;
		}
		else if(test==3) {
			if(!flush&&!suck) {
				cout << "bulk writing (" << keyval.first << "," << keyval.second << ")" << endl;
				wspace->write(keyval);

				if(partial_count++>=partial_limit) flush = true;
			}
			else if(flush&&!suck) {
				cout << "### flushing (k,v) pairs ----------------------------->" << endl;
				wspace->flush();

				flush = false;
				suck = true;
			}
			else {
				// take any key in local spaces
				keyval.first = "";

				wspace->takeIfExists(keyval);
				cout << "bulk taking (" << keyval.first << "," << keyval.second << ")" << endl;

				if(partial_count--<=0) suck = false;
			}
		}
		else if(test==4) {
			cout << "cleaning (" << keyval.first << "," << keyval.second << ")" << endl;
			wspace->clean(keyval.first);
			cout << "writing (" << keyval.first << "," << keyval.second << ")" << endl;
			wspace->write(keyval);
			wspace->readIfExists(keyval);
			cout << "reading (" << keyval.first << "," << keyval.second << ")" << endl;
		}
	}

	delete wspace;
}

void print_usage(char *prog)
{
	cout << "-------------------------------------------------------------------------" << endl;
	cout << "usage : " << prog << " [-c <config> or -p <port> -h <ip>] [-n <num>] [-1234 <test>]" << endl;
	cout << "\t-c\t\tconfiguration file" << endl;
	cout << "\t-p\t\tport" << endl;
	cout << "\t-h\t\tip to connect" << endl;
	cout << "\t-n\t\tprocess num" << endl;
	cout << "\t-1\t\twrite/read/take test in infinite loop" << endl;
	cout << "\t-2\t\twrite/read in some loop, and then take in some loop" << endl;
	cout << "\t-3\t\tbulk write test in infinite loop" << endl;
	cout << "\t-4\t\tclean and write in infinite loop" << endl;
	cout << endl;
	cout << "example : " << endl;
	cout << "-- space discovery server" << endl;
	cout << "\t" << prog << " -p 18088" << endl;
	cout << endl;
	cout << "-- space itself" << endl;
	cout << "\t" << prog << " -h localhost -p 18088 -n 2" << endl;
	cout << endl;
	cout << "-- space client test" << endl;
	cout << "\t" << prog << " -h localhost -p 18088 -1" << endl;
	cout << "-------------------------------------------------------------------------" << endl;
}

int main(int argc, char **argv)
{
	if(argc<2) {
		print_usage(argv[0]);
		exit(-1);
	}

	int opt;
	static char optstr[] = "cphn1234";

	char *conf= "coord.conf";
	int port = -1;
	int num = 1;
	string host = "";
	int test = -1;

	while((opt=getopt(argc,argv,optstr))!=-1) {
		switch(opt) {
			case 'c' :
			{
				conf = argv[optind++];
				CProperty config(conf);
				pair<string,string> kv;
                                kv.first = COORD_CONF_MASTER_IP_KEY;
                                host = config.get(kv).second;
                                kv.first = COORD_CONF_MASTER_PORT_KEY;
                                port = atoi((char *)config.get(kv).second.c_str());
				break;
			}
			case 'p' :
				port = atoi(argv[optind++]);
				break;
			case 'h' :
				host = argv[optind++];
				break;
			case 'n' :
				num = atoi(argv[optind++]);
				break;
			case '1' :
				test = 1;
				break;
			case '2' :
				test = 2;
				break;
			case '3' :
				test = 3;
				break;
			case '4' :
				test = 4;
				break;
			default :
				print_usage(argv[0]);
				exit(-1);
		}
	}

	if(port>0&&host=="") cube_sd(port);
	else if(port>0&&host!=""&&test>0) {
		//cube_sp_test(host, port, test);
		cube_sp_test2((char *)host.c_str(), port, test);
	}
	else {
		vector<pid_t> pids;
		for(int i=0;i<num;i++) {
			pid_t pid = fork();
			if(pid==0) { // child
				cube_sp((char *)host.c_str(), port);
				exit(1);
			}
			else if(pid>0) pids.push_back(pid); // parent
			else { // fork error
				sleep(1);
				i--;
			}
		}

		for(int i=0;i<num;i++) {
			waitpid(pids[i],NULL,0);
		}
	}

        return 1;
}
